package mq

import (
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"huijing_mall/pkg/logger"
	"huijing_mall/pkg/vipercfg"

	"sync"
	"time"
)

// RabbitMQ 连接管理器
type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	config  vipercfg.RabbitMQConfig
	mu      sync.Mutex // 用于保护通道操作的互斥锁
}

// NewRabbitMQ 初始化RabbitMQ连接
func NewRabbitMQ(config vipercfg.RabbitMQConfig) *RabbitMQ {
	// 连接RabbitMQ
	conn, err := amqp.Dial(config.URL)
	if err != nil {
		panic(fmt.Errorf("无法连接到RabbitMQ: %v", err))
	}

	// 创建通道
	ch, err := conn.Channel()
	if err != nil {
		conn.Close()
		panic(fmt.Errorf("无法创建通道: %v", err))
	}

	return &RabbitMQ{
		conn:    conn,
		channel: ch,
		config:  config,
	}
}

// Close 关闭连接
func (r *RabbitMQ) Close() error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.channel != nil {
		if err := r.channel.Close(); err != nil {
			logger.Info(fmt.Sprintf("关闭通道失败: %v", err))
		}
	}

	if r.conn != nil {
		return r.conn.Close()
	}
	return nil
}

// DeclareQueue 声明队列
func (r *RabbitMQ) DeclareQueue(queueName string) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.channel == nil {
		return errors.New("通道未初始化")
	}

	_, err := r.channel.QueueDeclare(
		queueName,
		r.config.Durable,
		r.config.AutoDelete,
		r.config.Exclusive,
		r.config.NoWait,
		nil,
	)
	return err
}

// Publish 发送消息到指定队列
func (r *RabbitMQ) Publish(queueName string, message []byte) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.channel == nil {
		return errors.New("通道未初始化")
	}

	return r.channel.Publish(
		"",        // 不使用交换机，直接发送到队列
		queueName, // 路由键设置为队列名
		false,     // mandatory
		false,     // immediate
		amqp.Publishing{
			ContentType:  "text/plain",
			Body:         message,
			DeliveryMode: amqp.Persistent, // 持久化消息
			Timestamp:    time.Now(),
		},
	)
}

func (r *RabbitMQ) Consume(queueName string, consumerID string, handler func([]byte) error) (func(wait bool), error) {
	if r.channel == nil {
		return nil, errors.New("通道未初始化")
	}

	// 获取消息通道（AMQP底层通道，非自定义通道）
	msgChan, err := r.channel.Consume(
		queueName,
		consumerID,
		false, // 手动Ack
		false, // 非排他
		false, // 不本地通知
		false, // 非阻塞（无消息时不阻塞）
		nil,
	)
	if err != nil {
		return nil, fmt.Errorf("注册消费者%s失败: %v", consumerID, err)
	}

	// 控制退出的核心通道（用缓冲通道避免goroutine泄漏，仅1个信号）
	stopChan := make(chan struct{}, 1)
	exitedChan := make(chan struct{}, 1)

	// 启动单个消费协程（每个消费者1个协程，低内存）
	go func() {
		defer func() {
			close(exitedChan) // 退出时通知
			logger.Info("消费者%s: 协程已退出", consumerID)
		}()
		logger.Info(fmt.Sprintf("消费者%s: 开始监听队列[%s]", consumerID, queueName))

		for {
			select {
			// 1. 正常接收消息
			case delivery, ok := <-msgChan:
				if !ok {
					logger.Warn(fmt.Sprintf("消费者%s: MQ消息通道已关闭（可能MQ断开）", consumerID))
					return
				}
				// 调用业务处理（带defer确保Ack/Nack不遗漏）
				handleDelivery(consumerID, delivery, handler)

			// 2. 收到停止信号
			case <-stopChan:
				logger.Info(fmt.Sprintf("消费者%s: 收到停止信号，准备退出", consumerID))
				// 主动取消MQ消费者注册（释放MQ服务器资源，避免内存泄漏）
				if err := r.channel.Cancel(consumerID, false); err != nil {
					logger.Error(fmt.Sprintf("消费者%s: 取消MQ注册失败: %v", consumerID, err))
				}
				return
			}
		}
	}()

	// 停止函数（仅在Ctrl+C时调用）
	stopFunc := func(wait bool) {
		stopChan <- struct{}{} // 发送停止信号（缓冲通道避免阻塞）
		if !wait {
			return
		}
		// 等待消费者完全退出（超时5秒，避免无限阻塞）
		select {
		case <-exitedChan:
			logger.Info(fmt.Sprintf("消费者%s: 已完全退出", consumerID))
		case <-time.After(5 * time.Second):
			logger.Warn(fmt.Sprintf("消费者%s: 等待退出超时（可能仍在处理消息）", consumerID))
		}
	}

	return stopFunc, nil
}

// handleDelivery 单独封装消息处理（解耦+确保Ack/Nack）
func handleDelivery(consumerID string, delivery amqp.Delivery, handler func([]byte) error) {
	// 无论处理成功/失败，都必须Ack/Nack（否则MQ消息会堆积）
	defer func() {
		if r := recover(); r != nil {
			logger.Error(fmt.Sprintf("消费者%s: 处理消息恐慌: %v, 消息: %s", consumerID, r, string(delivery.Body)))
			//_ = delivery.Nack(false, true) // 恐慌时重新入队
			delivery.Ack(false)
		}
	}()

	// 业务处理
	if err := handler(delivery.Body); err != nil {
		logger.Error(fmt.Sprintf("消费者%s: 处理消息失败: %v, 消息: %s", consumerID, err, string(delivery.Body)))
		//_ = delivery.Nack(false, true) // 失败重新入队
		delivery.Ack(false) //取消入队
		return
	}

	// 处理成功，确认消息
	if err := delivery.Ack(false); err != nil {
		logger.Error(fmt.Sprintf("消费者%s: 确认消息失败: %v", consumerID, err))
	}
}
